package com.example.kafka.controller;

import cn.hutool.json.JSONObject;
import com.example.base.result.AjaxResult;
import com.example.kafka.config.KafkaConfig;
import com.example.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * Kafka生产者
 * @author wanghongyu02
 * @date 2024-07-01 16:15
 */
@Slf4j
@RestController
@RequestMapping("/kafka/producer")
public class ProducerController {
    @Autowired
    KafkaTemplate<String, String> kafka;

    @Autowired
    KafkaConfig kafkaConfig;

    /**
     * 通过KafkaTemplate发送消息（带回调）
     * (默认读取yml配置文件中的配置，不推荐，因为配置文件中的配置是全局的，可能会和其他的地方出现冲突))
     * @author wanghongyu02
     * @date 2024-07-05 15:54
     * @param paramObj
     * @return com.example.base.result.AjaxResult
     */
    @PostMapping("/sendMsgByKafkaTemplate")
    public AjaxResult sendMsg(@RequestBody JSONObject paramObj) {
        String msg = paramObj.getStr("msg");

        if (StringUtils.isEmpty(msg)) {
            return AjaxResult.error("消息内容不能为空！");
        }
        log.info("接收到前端信息msg:{}", msg);

        try {
            log.info("接收到前端信息msg:{}", msg);

            kafka.send("mytopic1", msg).addCallback(
                    result -> log.info("消息发送成功: {}", msg),
                    ex -> log.error("消息发送失败: {}", msg, ex)
            );
        } catch (Exception e) {
            log.error("消息发送失败！msg:{}", msg, e);
        }

        return AjaxResult.success("消息发送请求已提交：" + msg);
    }

    /**
     * 通过KafkaProducer发送消息（手动读取配置文件）
     * @author wanghongyu02
     * @date 2024-07-05 15:54
     * @param paramObj
     * @return com.example.base.result.AjaxResult
     */
    @PostMapping("/sendMsgByKafkaProducer")
    public AjaxResult sendMsgByKafkaProducer(@RequestBody JSONObject paramObj) {

        // 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(kafkaConfig.getProducerProperties());

        // 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("mytopic1","kafkaProducerTest " + i));
        }

        // 关闭资源
        kafkaProducer.close();

        return AjaxResult.success("消息发送成功!");
    }

    /**
     * 通过KafkaProducer发送消息(带回调)
     * @author wanghongyu02
     * @date 2024-07-05 15:54
     * @param paramObj
     * @return com.example.base.result.AjaxResult
     */
    @PostMapping("/sendMsgWithCallback")
    public AjaxResult sendMsgWithCallback(@RequestBody JSONObject paramObj) throws InterruptedException {

        // 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(kafkaConfig.getProducerProperties());

        // 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            // 该方法在Producer收到ack时调用，为异步调用
            kafkaProducer.send(new ProducerRecord<>("mytopic1","kafkaProducerTest " + i), (metadata, e) -> {
                if (e == null) {
                    // 没有异常,输出信息到控制台
                    log.info("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition());
                } else {
                    // 出现异常打印
                    e.printStackTrace();
                }
            });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        // 关闭资源
        kafkaProducer.close();
        return AjaxResult.success("消息发送成功!");
    }
}
